/**
 * Copyright 2010 Creative Works, Inc.
 * Creative Works licenses this file to you under the Apache License, version
 * 2.0 (the "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *    http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied.  See the License for the specific language governing
 * permissions and limitations under the License.
 */
package org.sendvox.freswitch.inbound.impl;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
import org.sendvox.freswitch.core.Command;
import org.sendvox.freswitch.core.CommandException;
import org.sendvox.freswitch.core.NetAddress;
import org.sendvox.freswitch.event.impl.EventSocketMessageListener;
import org.sendvox.freswitch.event.impl.Message;
import org.sendvox.freswitch.event.impl.RawMessage;
import org.sendvox.freswitch.inbound.InboundClient;
import org.sendvox.freswitch.inbound.InboundException;
import org.sendvox.freswitch.util.Arguments;

/**
 * Simply put, the inbound client connects to the FreeSWITCH server to invoke
 * commands and control FreeSWITCH.
 * 
 */
public class InboundClientImpl extends InboundClient {

	private static final Logger logger = Logger
			.getLogger(InboundClientImpl.class.getName());

	private ClientBootstrap bootstrap;
	private ChannelFuture connectFuture;
	private ChannelGroup channelGroup;
	private InboundConnectionState state = InboundConnectionState.INITIAL;
	private Map<Option, String> configMap;
	public static final String SPACE = " ";
	private Channel inboundChannel;

	public void configurationOption(Map<Option, String> options) {
		Arguments.INSTANCE.validate(options).isNotNull();
		this.configMap = options;
	}

	private void init() {
		ChannelFactory factory = new NioClientSocketChannelFactory(
				Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool(), 3);

		// Set up.
		bootstrap = new ClientBootstrap(factory);

		channelGroup = new DefaultChannelGroup(
				InboundClientImpl.class.getName());

		// 200 threads max, Memory limitation: 1MB by channel, 1GB global, 100
		// ms of timeout
		OrderedMemoryAwareThreadPoolExecutor pipelineExecutor = new OrderedMemoryAwareThreadPoolExecutor(
				200, 1048576, 1073741824, 100, TimeUnit.MILLISECONDS,
				Executors.defaultThreadFactory());

		final int NUMB_CONNECTIONS = getConfigOption(Option.NUMB_CONNECTIONS);

		// Configure the event pipeline factory.
		bootstrap.setPipelineFactory(new InboundPipelineFactory(channelGroup,
				pipelineExecutor, null, NUMB_CONNECTIONS));

		// Add the handler to the pipeline and set some options
		bootstrap.setOption("tcpNoDelay", true);
		bootstrap.setOption("keepAlive", true);
		bootstrap.setOption("reuseAddress", true);

	}

	public void connect(NetAddress netAddress) throws InboundException {

		if (state != InboundConnectionState.INITIAL
				&& state != InboundConnectionState.DISCONNECTED) {
			throw new InboundException(
					"Connection may only be stablished when in state "
							+ "INITIAL or DISCONNECTED, but connection is in state "
							+ state);
		}

		this.init();

		logger.log(Level.INFO, "Connecting to " + netAddress.toString());

		state = InboundConnectionState.CONNECTING;

		// *** Start the Netty running ***

		// Make a new connection.
		connectFuture = bootstrap.connect(new InetSocketAddress(netAddress
				.getHost(), netAddress.getPort()));

		final int TIMEOUT_SECONDS = getConfigOption(Option.TIMEOUT_SECONDS);

		// Wait until the connection is made successfully.
		if (connectFuture.awaitUninterruptibly(TIMEOUT_SECONDS,
				TimeUnit.SECONDS)) {

			inboundChannel = connectFuture.getChannel();
			// Add the parent channel to the group
			channelGroup.add(inboundChannel);

			state = InboundConnectionState.CONNECTED;

			this.login(netAddress);

		} else {
			logger.log(Level.SEVERE,
					"Failed to connect to " + netAddress.toString());
			logger.log(Level.SEVERE, "  * reason: {}", connectFuture.getCause());

			state = InboundConnectionState.DISCONNECTED;

			throw new InboundException("Timeout connecting to "
					+ netAddress.toString());
		}

	}

	protected void login(NetAddress netAddress) throws InboundException {

		if (state == InboundConnectionState.CONNECTED) {
			// But may have failed anyway
			if (connectFuture != null && connectFuture.isSuccess()) {
				// Get the handler instance to initiate the request.
				AuthenticationChannelHandler authHandler = inboundChannel
						.getPipeline().get(AuthenticationChannelHandler.class);

				if (authHandler != null)
					authHandler.authenticationRequest(netAddress.getPassword());

				// Wait for the authentication handshake to call back
				while (!authHandler.isAuthenticated.get()) {
					try {
						Thread.sleep(250);
					} catch (InterruptedException e) {
						// ignore
					}
				}

				state = InboundConnectionState.AUTHENTICATED;
			}
		}

	}

	public boolean isConnected() {
		AuthenticationChannelHandler authHandler = inboundChannel.getPipeline()
				.get(AuthenticationChannelHandler.class);

		return inboundChannel != null && inboundChannel.isConnected()
				&& authHandler.isAuthenticated.get();
	}

	public RawMessage disconnect() throws CommandException {
		// Wait until the connection is closed or the connection attempt fails.
		// Shut down all thread pools to exit.
		if (isConnected()) {
			ChannelGroupFuture future = channelGroup.close();
			future.awaitUninterruptibly();
			bootstrap.releaseExternalResources();
			return execute("exit");
		}
		return null;
	}

	public String status() {
		return state.getValue();
	}

	/**
	 * Send an API command to the FreeSWITCH server. This method blocks further
	 * execution until the command has been executed.
	 * <p>
	 * <code>api($command, $args) is identical to execute("api $command", $args)</code>
	 * </p>
	 * <p>
	 * Reference: http://wiki.freeswitch.org/wiki/Event_Socket_Library#api
	 * </p>
	 * 
	 */
	public RawMessage apiExecuteCommand(Command command, Object... objects)
			throws CommandException {
		return execute(
				Command.API.getName().concat(SPACE).concat(command.getName()),
				objects);
	}

	/**
	 * <p>
	 * <code>bgapi($command[, $arguments][,$custom_job_uuid])</code>
	 * </p>
	 * Send a background API command to the FreeSWITCH server to be executed in
	 * it's own thread. This will be executed in it's own thread, and is
	 * non-blocking.
	 * <p>
	 * <code>bgapi($command, $args)</code> is identical to
	 * </code>execute("bgapi $command", $args)</code>
	 * </p>
	 * <p>
	 * Reference: http://wiki.freeswitch.org/wiki/Event_Socket_Library#bgapi
	 * </p>
	 */
	public RawMessage backgroundApiExecuteCommand(Command command,
			Object... objects) throws CommandException {
		return execute(
				Command.BGAPI.getName().concat(SPACE).concat(command.getName()),
				objects);
	}

	@Override
	public RawMessage execute(Command command, Object... objects)
			throws CommandException {
		return execute(command.getName(), objects);
	}

	@Override
	public RawMessage execute(String fullCommand) throws CommandException {
		if (connectFuture != null && connectFuture.isSuccess()
				&& state == InboundConnectionState.AUTHENTICATED) {
			// Get the handler instance to initiate the request.
			InboundChannelHandler handler = inboundChannel.getPipeline().get(
					InboundChannelHandler.class);

			if (handler != null)
				return handler.executeSyncCommand(inboundChannel, fullCommand);

		} else {
			throw new IllegalStateException(
					"Command may only be sent when in state "
							+ "AUTHENTICATED, but connection is in state "
							+ state);
		}
		return null;
	}

	private RawMessage execute(String command, Object... objects)
			throws CommandException {

		final String inputCommand = MessageFormatter.formatCommand(command,
				objects);

		return execute(inputCommand);
	}

	public InboundConnectionState getState() {
		return this.state;
	}

	@Override
	public void addMessageHandler(EventSocketMessageListener<Message> listener) {
		if (connectFuture != null && connectFuture.isSuccess()) {
			Channel channel = connectFuture.getChannel();
			// Get the handler instance to initiate the request.
			InboundChannelHandler handler = channel.getPipeline().get(
					InboundChannelHandler.class);
			handler.addEventSocketListener(listener);
		}
	}

	@Override
	public void deleteMessageHandler(
			EventSocketMessageListener<Message> listener) {
		if (connectFuture != null && connectFuture.isSuccess()) {
			Channel channel = connectFuture.getChannel();
			// Get the handler instance to initiate the request.
			InboundChannelHandler handler = channel.getPipeline().get(
					InboundChannelHandler.class);
			handler.removeEventSocketListener(listener);
		}
	}

	private Integer getConfigOption(Option option) {
		int intValue = Integer.MAX_VALUE;
		if (configMap != null && configMap.containsKey(option)) {
			intValue = Integer.parseInt(configMap.get(option));
		}
		return intValue;
	}

	private enum InboundConnectionState {

		INITIAL("INITIAL"), CONNECTING("CONNECTING"), CONNECTED("CONNECTED"), AUTHENTICATED(
				"AUTHENTICATED"), RECONNECTING("RECONNECTING"), DISCONNECTING(
				"DISCONNECTING"), DISCONNECTED("DISCONNECTED");

		private final String value;

		private InboundConnectionState(String value) {
			this.value = value;
		}

		public String getValue() {
			return value;
		}

	}

	public InboundClient getThis() {
		return this;
	}

}
